package com.se.nju.rtvps.service;


import net.jxta.endpoint.Message;
import net.jxta.peergroup.PeerGroup;
import net.jxta.pipe.PipeMsgEvent;
import net.jxta.pipe.PipeMsgListener;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.util.JxtaBiDiPipe;
import net.jxta.util.JxtaServerPipe;
import com.se.nju.rtvps.main.ServiceConstants;
import com.se.nju.rtvps.operation.Operation;

public class ControlServiceThread implements Runnable{

	private PipeAdvertisement pipeAdv;
	private OperationFactory opFactory;
	private PeerGroup pg;
	
	public ControlServiceThread(PeerGroup pg, PipeAdvertisement pipeAdv, OperationFactory opFactory){
		this.pg = pg;
		this.pipeAdv = pipeAdv;
		this.opFactory = opFactory;
	}
	public void run() {
		System.out.println("start pipe listenning for control service~~~~~~");
		try {
			JxtaServerPipe serverPipe = new JxtaServerPipe(pg, pipeAdv );
			serverPipe.setPipeTimeout(0);
			while(true){
				JxtaBiDiPipe bdpipe = serverPipe.accept();
				if(bdpipe != null){
					Thread thread = new Thread(new MsgListener(bdpipe));
					thread.start();
				}
			}
		} catch (Exception e) {
			System.out.println("Control Service Listening Thread Error!!");
			e.printStackTrace();
		} 
			

	}
	
	class MsgListener implements Runnable, PipeMsgListener{

		private JxtaBiDiPipe bdpipe;
		
		public MsgListener(JxtaBiDiPipe bdpipe){
			this.bdpipe = bdpipe;
		}
		public void run() {
			bdpipe.setMessageListener(this);
			while(bdpipe.isBound()){}
		}

		public void pipeMsgEvent(PipeMsgEvent event) {
			Message msg = event.getMessage();
			System.out.println("Msg Received=========================\n"+msg);
			//String tagName = msg.getMessageElement(ServiceConstants.OPERATION_TAG).toString();
			//Operation operation = opFactory.getOperation(tagName);
			//operation.execute(msg, bdpipe);
		}
	}

}
 